/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package comunicacao;

import entities.Server;
import java.io.IOException;
import java.net.DatagramSocket;
import java.net.ServerSocket;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.swing.JTextArea;
import main.ConfigGeral;
import main.DataWatcherRunnable;
import org.apache.log4j.ConsoleAppender;
import org.apache.log4j.PatternLayout;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

/**
 *
 * @author Leonardo Lino Nº45340
 * @author Diogo Pires Nº40351
 */
public class GestorZooKeeper {

    private ZooKeeper zk;
    private String thisAddress;
    private int thisPort;
    private JTextArea consola;
    private String znodeName;

    public GestorZooKeeper(String thisAddress, int thisPort, JTextArea consola) {
        this.consola = consola;
        this.thisAddress = thisAddress;
        this.thisPort = thisPort;

        try {
            ConsoleAppender console = new ConsoleAppender(); //create appender
            //configure the appender
            String PATTERN = "%d [%p|%c|%C{1}] %m%n";
            console.setLayout(new PatternLayout(PATTERN));
            console.setThreshold(org.apache.log4j.Level.ERROR);
            console.activateOptions();
            org.apache.log4j.BasicConfigurator.configure(console);
            
            connect(ConfigGeral.CG_ZOOKEEP_IP + ":" + ConfigGeral.CG_ZOOKEEP_PORT, 455);
        } catch (IOException ex) {
            //Logger.getLogger(GestorZooKeeper.class.getName()).log(Level.SEVERE, null, ex);
            System.err.println(ex.toString());
        } catch (InterruptedException ex) {
            // Logger.getLogger(GestorZooKeeper.class.getName()).log(Level.SEVERE, null, ex);
            System.err.println(ex.toString());
        }
    }

    public Server getPrimary() {
        Server server = null;

        try {
            Stat stat = zk.exists("/Ready/DBready", null);
            byte[] dados = zk.getData("/Ready/DBready", null, stat);

            String dadosStr = new String(dados);
            String[] piece = dadosStr.split(",");
            String[] segundo = piece[0].split(":");
            server = new Server(segundo[0], Integer.parseInt(segundo[1]));
            server.setRmiBindName(segundo[2]);
            
            

        } catch (KeeperException ex) {
            Logger.getLogger(GestorZooKeeper.class.getName()).log(Level.SEVERE, null, ex);
            consola.append("ERRO: " + ex.toString());
        } catch (InterruptedException ex) {
            Logger.getLogger(GestorZooKeeper.class.getName()).log(Level.SEVERE, null, ex);
            consola.append("ERRO: " + ex.toString());
        }
        return server;
    }

    public void start() {
        try {
            Stat ready = zk.exists("/Ready", null);

            consola.append("entrei start \n");

            if (ready == null) {
                zk.create("/Ready",
                        null,
                        ZooDefs.Ids.OPEN_ACL_UNSAFE,
                        CreateMode.PERSISTENT);
            }

            Stat appServers = zk.exists("/AppServers", null);

            if (appServers == null) {
                initialSetup();
            } else {
                addNewServer();
            }

            DataWatcherRunnable data = new DataWatcherRunnable(zk);
            data.run();

            list("AppServers");

        } catch (KeeperException ex) {
            Logger.getLogger(GestorZooKeeper.class.getName()).log(Level.SEVERE, null, ex);
            consola.append("ERRO: " + ex.toString());
        } catch (InterruptedException ex) {
            Logger.getLogger(GestorZooKeeper.class.getName()).log(Level.SEVERE, null, ex);
            consola.append("ERRO: " + ex.toString());
        }
    }

    private void addNewServer() throws KeeperException, InterruptedException {

        int counter = 0;
        Stat stat = zk.exists("/AppServersConfig/Counter", null);
        byte[] dados = zk.getData("/AppServersConfig/Counter", null, stat);

        Stat appReady = zk.exists("/Ready/Appready", null);
        if (appReady == null) {
            byte[] data2 = (thisAddress + ":" + thisPort).getBytes();
            create("Ready/Appready", data2);
            counter = 1;

        } else {

            String s = new String(dados);
            counter = Integer.parseInt(s);
            counter++;
        }

        znodeName = "App" + counter;
        byte[] dataArray = (thisAddress + ":" + thisPort).getBytes();
        create("AppServers/" + znodeName, dataArray);

        zk.setData("/AppServersConfig/Counter", Integer.toString(counter).getBytes(), -1);

    }

    private void initialSetup() throws KeeperException, InterruptedException {


        zk.create("/AppServers",
                null,
                ZooDefs.Ids.OPEN_ACL_UNSAFE,
                CreateMode.PERSISTENT);

        /*
        zk.create("/AppServersConfig",
                null,
                ZooDefs.Ids.OPEN_ACL_UNSAFE,
                CreateMode.PERSISTENT);
        */
        znodeName = "App1";

        byte[] dataArray = (thisAddress + ":" + thisPort).getBytes();
        create("AppServers/" + znodeName, dataArray);

        byte[] count = Integer.toString(1).getBytes();

        zk.create("/AppServersConfig/Counter",
                count,
                ZooDefs.Ids.OPEN_ACL_UNSAFE,
                CreateMode.PERSISTENT);

        Stat appReady = zk.exists("/Ready/Appready", null);
        if (appReady == null) {

            byte[] data2 = (thisAddress + ":" + thisPort).getBytes();

            create("Ready/Appready", data2);
        }

    }

    public ZooKeeper connect(String hosts, int timeout)
            throws IOException, InterruptedException {

        final CountDownLatch signal = new CountDownLatch(1);

        zk = new ZooKeeper(hosts, timeout, null);
        return zk;
    }

    public void create(String groupName, byte[] dataArray)
            throws KeeperException, InterruptedException {

        String path = "/" + groupName;
        String createdPath = zk.create(path,
                dataArray,
                ZooDefs.Ids.OPEN_ACL_UNSAFE,
                CreateMode.EPHEMERAL);

        System.out.println("Created " + createdPath);

    }

    public void list(String groupName)
            throws KeeperException, InterruptedException {

        String path = "/" + groupName;

        try {
            List<String> children = zk.getChildren(path, false);
            for (String child : children) {
                System.out.println(child);
                consola.append(child + "\n");
            }
        } catch (KeeperException.NoNodeException e) {
            System.out.printf("Group %s does not exist \n", groupName);
        }
    }

    public String getZnodeName() {
        return znodeName;
    }

}
